package com.zhang.gmall.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.zhang.gmall.bean.{DauInfo, PageLog}
import com.zhang.gmall.util.{JedisUtil, MyBeanUtil, MyEsUtils, MyKafkaUtils, MyOffsetUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.{Jedis, Pipeline}

import java.time.{LocalDate, Period}
import java.lang
import java.text.SimpleDateFormat
import java.util.Date
import scala.collection.mutable.ListBuffer

/**
 * @title: 日活宽表
 * @author: zhang
 * @date: 2022/3/22 16:51 
 */
object DWDDauApp {
  def main(args: Array[String]): Unit = {

    //TODO 1.准备实时环境
    val sparkConf: SparkConf = new SparkConf().setAppName("DWDDauApp").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    //TODO 2.从redis中读取偏移量
    val topic = "DWD_PAGE_LOG_TOPIC_SPARK"
    val groupId = "DWS_DAU_GROUP"
    val offsets: Map[TopicPartition, Long] = MyOffsetUtil.getOffset(topic, groupId)
    var kafkaDS: InputDStream[ConsumerRecord[String, String]] = null
    if (offsets != null && offsets.nonEmpty) {
      kafkaDS = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId, offsets)
    } else {
      kafkaDS = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId)
    }

    //TODO 3.提取偏移量结束点
    var offsetRanges: Array[OffsetRange] = null
    val offsetRangesDS: DStream[ConsumerRecord[String, String]] = kafkaDS.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )


    //TODO 4.转换数据结构
    val pageLogDS: DStream[PageLog] = offsetRangesDS.map(
      data => {
        val pageLog: PageLog = JSON.parseObject(data.value(), classOf[PageLog])
        pageLog
      }
    )


    //TODO 5.对页面访问数据去重自我检=
    val filterDS: DStream[PageLog] = pageLogDS.filter(
      pageLog => pageLog.last_page_id == null
    )

    //TODO 6.借助redis对数据进一步去重
    val redisFilterDS: DStream[PageLog] = filterDS.mapPartitions(
      pageLogIter => {
        //获取redis连接
        val jedis: Jedis = JedisUtil.getJedis()
        val filterList: ListBuffer[PageLog] = ListBuffer[PageLog]()
        val pageLogList: List[PageLog] = pageLogIter.toList
        println("第三方审查前：" + pageLogList.size)
        val sdf = new SimpleDateFormat("yyyy-MM-dd")
        for (pageLog <- pageLogList) {
          val date: String = sdf.format(new Date(pageLog.ts))
          val redisKey = s"DAU:$date"
          val isNew: lang.Long = jedis.sadd(redisKey, pageLog.mid)
          if (isNew == 1L) {
            jedis.expire(redisKey, 60 * 60 * 24)
            filterList.append(pageLog)
          }
        }
        jedis.close()
        println("第三方审查后：" + filterList.size)
        filterList.toIterator
      }
    )
    //TODO 7.关联维度数据
    val dauInfoDS: DStream[DauInfo] = redisFilterDS.mapPartitions(

      pageLogIter => {
        val dauInfos: ListBuffer[DauInfo] = ListBuffer[DauInfo]()
        val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        val jedis: Jedis = JedisUtil.getJedis()
        for (pageLog <- pageLogIter) {
          val dauInfo = new DauInfo()
          //将pageLog信息copy到DauInfo中
          MyBeanUtil.copyProperties(pageLog, dauInfo)
          //TODO 7.1 补充用户维度
          val userId: String = pageLog.user_id
          val redisUserIdKey = s"DIM:USER_INFO:$userId"
          val userInfoJsonStr: String = jedis.get(redisUserIdKey)
          val userJSONObj: JSONObject = JSON.parseObject(userInfoJsonStr)
          //提取性别信息,并设置到DauInfo
          dauInfo.user_gender = userJSONObj.getString("gender")
          //提取生日,计算年龄
          val birthday: String = userJSONObj.getString("birthday")
          val birthdayLd: LocalDate = LocalDate.parse(birthday)
          val nowLd: LocalDate = LocalDate.now()
          val period: Period = Period.between(birthdayLd, nowLd)
          val age: Int = period.getYears
          dauInfo.user_age = age.toString
          //TODO 7.2 补充地区维度
          val provinceId: String = pageLog.province_id
          val redisProvinceIdKey = s"DIM:BASE_PROVINCE:$provinceId"
          val provinceIdStr: String = jedis.get(redisProvinceIdKey)
          val provinceIdObj: JSONObject = JSON.parseObject(provinceIdStr)
          dauInfo.province_name = provinceIdObj.getString("name")
          dauInfo.province_iso_code = provinceIdObj.getString("iso_code")
          dauInfo.province_3166_2 = provinceIdObj.getString("iso_3166_2")
          dauInfo.province_area_code = provinceIdObj.getString("area_code")
          //TODO 7.3 补充时间维度
          val date: String = sdf.format(new Date(pageLog.ts))
          val dateArr: Array[String] = date.split(" ")
          val dt = dateArr(0)
          val hr = dateArr(1).split(":")(0)
          dauInfo.dt = dt
          dauInfo.hr = hr
          dauInfos.append(dauInfo)
        }

        jedis.close()
        dauInfos.iterator
      }
    )

    //TODO 8.写入OLAP
    dauInfoDS.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          dauInfoIter => {
            val docs: List[(String, DauInfo)] = dauInfoIter.map(dauInfo => (dauInfo.mid, dauInfo)).toList
            if (docs.size > 0) {
              val head: (String, DauInfo) = docs.head
              val ts: Long = head._2.ts
              val sdf = new SimpleDateFormat("yyyy-MM-dd")
              val dateStr: String = sdf.format(new Date(ts))
              val indexName = s"gmall_dau_info_$dateStr"
              MyEsUtils.bulkSaveIdempotent(docs, indexName)
            }
          }

        )
        //提交offset
        MyOffsetUtil.saveOffset(topic, groupId, offsetRanges)
      }
    )
    //TODO 1.
    //TODO 9.启动任务
    ssc.start()
    ssc.awaitTermination()
  }

  //状态还原，在每次启动时候进行一次状态还原，以ES为准，将所有的mid提取出来，覆盖到redis中
  def resetState(): Unit = {
    //从es中查mid
    val date: LocalDate = LocalDate.now()
    val indexName: String = s"gmall_dau_info_$date"
    val fieldName = "mid"
    val mids: List[String] = MyEsUtils.searchField(indexName, fieldName)
    //不管mids是否有数据都要 删除redis中的记录状态
    val jedis: Jedis = JedisUtil.getJedis()
    val redisKey = s"DAU:$date"
    jedis.del(redisKey)
    //如果mids有数据，则覆盖redis中记录数据
    if (mids != null && mids.size > 0) {
      val pipeline: Pipeline = jedis.pipelined()
      for (mid <- mids) {
        pipeline.sadd(redisKey, mid) //不会直接写入redis
      }
      pipeline.sync() //写到redis中
    }
    jedis.close()
  }


}
